feat(phaser): add reusable phase coordination - #250
Conversation
There was a problem hiding this comment.
🟡 Changes recommended
Phaser::poll_wait currently clones wakers unconditionally instead of following the repo’s established WaitSet::will_wake pattern, adding avoidable per-poll overhead.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Pull request overview
Adds a new opt-in asyncband::phaser synchronization primitive for coordinating repeated phases with a dynamic participant set, including RAII participants, cancel-safe waiting semantics, and documentation/tests to integrate it into the crate’s public surface.
Changes:
- Introduce
asyncband::phaser::{Phase, Phaser, PhaserParticipant}behind a newphaserfeature flag. - Add unit + integration tests covering registration/arrival/advance semantics and cross-task waiting.
- Update crate docs (README + crate-level docs) and changelog to advertise the new primitive.
File summaries
| File | Description |
|---|---|
| tests-integration/tests/traits_test.rs | Extends trait assertions (Send/Sync/Unpin) to cover new public phaser types. |
| tests-integration/tests/phaser_test.rs | Adds integration tests for spawned-task waiting and observer semantics. |
| tests-integration/Cargo.toml | Enables asyncband’s new phaser feature for integration tests. |
| README.md | Documents the new Phaser feature in the public feature table. |
| CHANGELOG.md | Records the addition of the new opt-in Phaser. |
| Cargo.lock | Captures new dev/test dependency resolution (e.g., tokio-test). |
| asyncband/src/phaser/tests.rs | Adds focused unit tests for phase advancement, cancellation, waker behavior, and wraparound. |
| asyncband/src/phaser/mod.rs | Implements the new Phaser primitive, participants, waiting, and internal state transitions. |
| asyncband/src/lib.rs | Wires the phaser module into the crate behind a feature flag and updates crate docs. |
| asyncband/src/internal/mod.rs | Extends internal module feature gating to include phaser where needed. |
| asyncband/Cargo.toml | Adds the phaser feature and includes tokio-test for module tests. |
Review details
- Files reviewed: 10/11 changed files
- Comments generated: 1
- Review effort level: Lite
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| fn poll_wait( | ||
| &self, | ||
| token: &mut Option<WakerToken>, | ||
| observed: Phase, | ||
| cx: &mut Context<'_>, | ||
| ) -> Poll<Phase> { | ||
| let waker = cx.waker().clone(); | ||
| let _retired_waker = { | ||
| let mut state = self.state.lock(); | ||
| if state.phase != observed { | ||
| let phase = state.phase; | ||
| *token = None; | ||
| return Poll::Ready(phase); | ||
| } | ||
| state.waiters.register(token, waker) | ||
| }; | ||
| Poll::Pending | ||
| } |
orthur2
left a comment
orthur2
left a comment
There was a problem hiding this comment.
Could you rebase this onto current main branch first? Since the branch was last updated, #273 replaced WaitSet with WakerSet, and the recent API-table changes now conflict with this branch.
I'd also recommend adding a short Summary to the PR description, in line with AGENTS.md.You can refer to the body of PRs that have already been merged.
I'd be happy to do a more thorough review once it's rebased.
|
Updated. |
| //! | ||
| //! Waiters compare phase identity instead of inferring transitions from party counts. | ||
| //! | ||
| //! # Examples |
There was a problem hiding this comment.
Could we add an async example showing a few phases, including separate arrival and waiting?
There was a problem hiding this comment.
Currently, a spawned task needs a separate Arc<Phaser> to call wait_for_advance() after participant.arrive(). The participant already holds that same Arc internally. I'd suggest exposing a phaser() accessor so callers can use the two operations separately without carrying another handle.
| //! | ||
| //! [`Phaser::arrived_parties`] is the difference between the registered and unarrived counts. | ||
| //! | ||
| //! All state transitions and waiter registration share one synchronization point. |
There was a problem hiding this comment.
Could we spell out the memory visibility guarantee here?
The shared synchronization point explains how the internal transitions are ordered, but does not explicitly tell callers whether operations before each participant's arrival happen-before operations after a wait observes that phase's completion. The shared mutex already provides this guarantee, but it should be explicit in the public API docs.
I added a # Synchronization section for this in ManualResetEvent last week. Something similar would help here.
|
Thanks for the review. I've addressed the comments above. |
Signed-off-by: tison <wander4096@gmail.com>
Register under the existing state lock and defer replaced waker destruction until after unlocking, following the contract from apache#257. Remove the separate probe and owned registration APIs, and replace clone-reentrancy coverage with replacement and cancellation drop-reentrancy tests.
tisonkun
left a comment
tisonkun
left a comment
There was a problem hiding this comment.
Thanks for your contribution! I reuse your codebase and add some bench baseline and modify the API a bit.
Welcome to drop a review after merge and if anything can be improved, feel free to submit a new PR.
|
Thanks for the refinements. The changes look great. I’ll follow up if I find anything that can be improved. |
|
@ButterBright Improvements can be driven by examples - I see Java Phaser can have parent but the callback style is not like Rust idiom. You may take a look at other primitives or see if there is common performance tricks (esp. for channels) |
Summary
Phaserfor repeated rounds with dynamic, owned participants.register_one()registers one participant;register(n)atomically registers a batch and returns an owning iterator.wait(), splitarrive()/wait(), independentPhaser::wait(observed)observers, and explicitclose()with an opaqueClosederror.Design Notes
A participant owns one arrival obligation per phase. Repeated arrivals in that phase count once. Cancelling a polled participant wait preserves its committed arrival and pending phase, so retrying observes the same round. Dropping a participant or the unconsumed portion of a batch withdraws those registrations; withdrawal does not certify successful application work. Applications can close the phaser on failure.
Empty phasers remain dormant and reusable. Closing freezes the unfinished phase and releases its waiters with
Closed; already completed phases remain successful. Phase numbers are wrappingu64observations: waiting detects a change, not arrival at a future numeric target.Waker registration uses the existing borrowed-waker API inside one state critical section, following the contract adopted in #257. Repeated polls reuse a matching registration; replacement cleanup and notification run outside the lock. The examples express grouped coordination and asynchronous work between rounds by composing phasers and coordinator tasks.